package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

import scala.reflect.ClassTag

/**
  * Created by Administrator on 2017/5/4 0004.
  * 图同构
  */
object TestGraphIsomorphism {

  def readRdf(sc:SparkContext, filename:String) ={
    val r = sc.textFile(filename).map(_.split("\t"))
    //使用与mergeGraphs函数类似的方法来创建一个包含不重复顶点的字典。
    val v = r.map(_(1)).union(r.map(_(3))).distinct.zipWithIndex
    //另一个借鉴mergeGraphs的编码技巧：通过交换元素操作，将存储顶点名称与ID关系的RDD转换成VertexRDD.
    Graph(v.map(_.swap),
      r.map(x => (x(1),(x(2),x(3))))
        .join(v)
        .map(x => (x._2._1._2,(x._2._2, x._2._1._1)))
        .join(v)
        .map(x => new Edge(x._2._1._1, x._2._2, x._2._1._2)))
  }

  def removeSingletons[VD:ClassTag, ED:ClassTag](g:Graph[VD,ED]) =
    Graph(g.triplets.map(et =>(et.srcId, et.srcAttr))
      .union(g.triplets.map(et => (et.dstId, et.dstAttr)))
      .distinct, g.edges)

  //为什么要实现pred的另一种实现方式。因为我们想在map()的括号内使用pred，而在这种情况下，我们就不可以传递rdd或者图对象给pred作参数了。
  //所以我们要使用另外一种pred的实现方式。将要传递的rdd或者图转化为Scala Map。自定义的pred接收Scala Map类型的参数。
  def pred(v:Map[VertexId,(Array[Double], Array[Double], Double, Double)], mean:Double, u:Long, i:Long) = {
    val user = v.getOrElse(u, (Array(0.0), Array(0.0), 0.0, 0.0))
    val item = v.getOrElse(i, (Array(0.0), Array(0.0), 0.0, 0.0))
    mean + user._3 + item._3 + item._1.zip(user._2).map(x => x._1*x._2).reduce(_ + _)
  }

  //将图转换为Scala Map，好将该数值传递给自定义的pred。
  def vertexMap(g:Graph[(Array[Double], Array[Double], Double, Double), Double]) = g.vertices.collect.map(v => v._1 -> v._2).toMap

  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    conf.set("spark.driver.memory","4g")
    conf.set("spark.driver.cores","8")
    val sc = new SparkContext(conf)


    //在YAGO的"<exports>"边集上训练SVDPlusPlus模型
    val gf = readRdf(sc, "data/yagoFacts.tsv").subgraph(_.attr == "<exports>")
    val e = gf.edges.map(e => Edge(e.srcId, e.dstId, 1.0))
    //下面一行Conf(...)中的第四个参数是5,表示最高评分是5.然而我们这里的情境下，只有0分和1分，所以，我们就把5改为1也许更好。
    //当然了，我还会去试试5的。
    val (gs, mean) = lib.SVDPlusPlus.run(e, new lib.SVDPlusPlus.Conf(2, 10, 0, 5, 0.007, 0.007, 0.005, 0.015))

    //计算vr，缺失的加拿大潜在出口商品列表。
    val gc = removeSingletons(gf.subgraph(et => et.srcAttr == "<Canada>"))
    //vr是指在维基百科里，所有的加拿大没有出口，而其它国家有出口的商品的ID（顶点ID）。
    val vr = e.map(x => (x.dstId, ""))
              .distinct
              .subtractByKey(gc.vertices)
              .map(_._1)

    //查找没有记录在Wikipedia上的加拿大最有可能的出口商品。
    val vm = vertexMap(gs)
    //获取加拿大的ID（顶点ID）
    val cid = gf.vertices.filter(_._2 == "<Canada>").first._1
    //r是从vr变化得来，vr是rdd，所以r也是rdd。r中的每个元素都是tuple2.这个tuple2中的第一个元素是商品ID，第二个元素是分数（分数越
    // 高，表示加拿大事实上出口它的可能性越高。）
    val r = vr.map( v => (v, pred(vm, mean, cid, v)))

    //根据分值（分值是tuple2的第2个元素）求最大的tuple2，只返回第一个元素（商品id/顶点ID）
    val maxKey = r.max()(new Ordering[Tuple2[VertexId,Double]](){
      override def compare(x: (VertexId, Double), y: (VertexId, Double)) : Int =
        Ordering[Double].compare(x._2, y._2)
    })._1

    val xxx = gf.vertices.filter(_._1 == maxKey).collect
    for(x <- xxx){
      println(x._1+"——"+x._2)
    }
    sc.stop
  }
}
